package cn.iocoder.yudao.module.bpm.message.kafka;

import cn.iocoder.yudao.module.bpm.message.ProcessInstanceMessage;
import cn.iocoder.yudao.module.bpm.message.dto.*;
import lombok.extern.log4j.Log4j2;
import org.springframework.kafka.core.KafkaTemplate;

import java.util.concurrent.ExecutionException;

@Log4j2
public class KafkaProcessInstanceMessage implements ProcessInstanceMessage {

    private final KafkaTemplate<Object, Object> kafkaTemplate;

    private final String processInstanceCleanTopic;
    private final String processInstanceFinishedTopic;
    private final String processInstanceRejectTopic;
    private final String processInstanceCreatedTopic;
    private final String processInstanceStarted;

    public KafkaProcessInstanceMessage(
            KafkaTemplate<Object, Object> kafkaTemplate,
            String processInstanceCleanTopic,
            String processInstanceFinishedTopic,
            String processInstanceRejectTopic,
            String processInstanceCreatedTopic,
            String processInstanceStarted
    ) {
        this.kafkaTemplate = kafkaTemplate;
        this.processInstanceCleanTopic = processInstanceCleanTopic;
        this.processInstanceFinishedTopic = processInstanceFinishedTopic;
        this.processInstanceRejectTopic = processInstanceRejectTopic;
        this.processInstanceCreatedTopic = processInstanceCreatedTopic;
        this.processInstanceStarted = processInstanceStarted;
    }
    @Override
    public void onProcessInstanceCleanMsg(BpmProcessInstanceCancelledMessage bpmProcessInstanceCancelledMessage) {
        try {
            kafkaTemplate.send(processInstanceCleanTopic, null, bpmProcessInstanceCancelledMessage).get();
        } catch (InterruptedException | ExecutionException e) {
            log.error("[sendKafkaMessage][发送消息({}) 到 Kafka 失败]", bpmProcessInstanceCancelledMessage, e);
        }
    }

    @Override
    public void onProcessInstanceFinishedMsg(BpmProcessInstanceApproveMessage bpmProcessInstanceApproveMessage) {
        try {
            kafkaTemplate.send(processInstanceFinishedTopic, null, bpmProcessInstanceApproveMessage).get();
        } catch (InterruptedException | ExecutionException e) {
            log.error("[sendKafkaMessage][发送消息({}) 到 Kafka 失败]", bpmProcessInstanceApproveMessage, e);
        }
    }

    @Override
    public void onProcessInstanceRejectMsg(BpmProcessInstanceRejectMessage bpmProcessInstanceRejectMessage) {
        try {
            kafkaTemplate.send(processInstanceRejectTopic, null, bpmProcessInstanceRejectMessage).get();
        } catch (InterruptedException | ExecutionException e) {
            log.error("[sendKafkaMessage][发送消息({}) 到 Kafka 失败]", bpmProcessInstanceRejectMessage, e);
        }
    }

    @Override
    public void onProcessInstanceCreatedMsg(BpmProcessInstanceCreatedMessage bpmProcessInstanceCreatedMessage) {
        try {
            kafkaTemplate.send(processInstanceCreatedTopic, null, bpmProcessInstanceCreatedMessage).get();
        } catch (InterruptedException | ExecutionException e) {
            log.error("[sendKafkaMessage][发送消息({}) 到 Kafka 失败]", bpmProcessInstanceCreatedMessage, e);
        }
    }

    @Override
    public void onProcessInstanceStartedMsg(BpmProcessInstanceStartedMessage bpmProcessInstanceStartedMessage) {
        try {
            kafkaTemplate.send(processInstanceStarted, null, bpmProcessInstanceStartedMessage).get();
        } catch (InterruptedException | ExecutionException e) {
            log.error("[sendKafkaMessage][发送消息({}) 到 Kafka 失败]", bpmProcessInstanceStartedMessage, e);
        }
    }
}
